package org.apache.flink.kubernetes.kubeclient.decorators;

import org.apache.flink.kubernetes.configuration.KubernetesDeploymentTarget;
import org.apache.flink.kubernetes.kubeclient.ComponentPod;
import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesMasterComponentParameters;
import org.apache.flink.kubernetes.utils.Constants;
import org.apache.flink.kubernetes.utils.KubernetesUtils;

import io.fabric8.kubernetes.api.model.Container;
import io.fabric8.kubernetes.api.model.ContainerBuilder;

import java.util.List;

import static org.apache.flink.util.Preconditions.checkNotNull;

/** Attach the command and args to the main container for running the ConsumerManager. */
/**
 * @类名: CmdConsumerManagerDecorator
 * @说明: 
 *
 * @author   leehom
 * @Date	 2024年4月3日 下午1:47:26
 * 修改记录：
 *
 * @see 	 
 */
public class CmdConsumerManagerDecorator extends AbstractKubernetesStepDecorator {

    private final KubernetesMasterComponentParameters kubernetesConsumerManagerParameters;

    public CmdConsumerManagerDecorator(KubernetesMasterComponentParameters kubernetesJobManagerParameters) {
        this.kubernetesConsumerManagerParameters = checkNotNull(kubernetesJobManagerParameters);
    }

    @Override
    public ComponentPod decoratePod(ComponentPod componentPod) {
        final Container mainContainerWithStartCmd =
                new ContainerBuilder(componentPod.getMainContainer())
                        .withCommand(getConsumerManagerStartCommand())
                        .build();

        return new ComponentPod.Builder(componentPod).withMainContainer(mainContainerWithStartCmd).build();
    }

    private List<String> getConsumerManagerStartCommand() {
        final KubernetesDeploymentTarget deploymentTarget =
                KubernetesDeploymentTarget.fromConfig(
                        kubernetesConsumerManagerParameters.getConfiguration());
        // 构建作业管理器启动脚本
        return KubernetesUtils.getStartCommandWithBashWrapper(
                Constants.KUBERNETES_CONSUMER_MANAGER_SCRIPT_PATH
                        + " "
                        + deploymentTarget.getName()
                        + " "
                        + kubernetesConsumerManagerParameters.getEntrypointArgs());
    }
}
